package com.zhao.biz.offline.sparksql

import java.text.SimpleDateFormat

import com.zhao.common.CommonData
import com.zhao.dao.impl.GameAnaylysisResultImpl
import com.zhao.entity.GameAnaylysisResultBean
import com.zhao.utils.CommonUtils
import org.apache.spark.{SPARK_BRANCH, SparkContext}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

/**
 * Description: 离线统计(新增用户,活跃用户,次日留存率~7日留存率)<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/17 14:35
 *
 * @author 柒柒
 * @version : 1.0
 */

object HotBloodOfflineCalSQL {
  def main(args: Array[String]): Unit = {
    //步骤:
    //1.拦截非法的参数
    if (args == null || args.length != 1){
      println(
        """
          |警告!
          |请录入参数 <基准日>,如:2018-02-01
          |""".stripMargin)
      sys.exit(-1)
    }

    //2.获得参数(基准日)
    val Array(baseDate) = args

    //3.SparkSession
    val (spark, sc) = getSparkContext

    //4.从ES中读取数据,并cache虚拟表
    readDataFromES(spark,sc)

    //5.分别计算指标
    //注册自定义函数
    spark.udf.register("getTimeMills",(timeStr: String) =>processTime(timeStr))

    val row: Row = spark.sql(
      s"""
         |select
         |    max(newUser) `新增用户数`,
         |    max(activeUser) `活跃用户数`,
         |    max(nextDayStayCnt) * 100.0 / max(newUsers) `次日留存率`,
         |    max(twoDayStayCnt) * 100.0 / max(newUsers) `2日留存率`,
         |    max(threeDayStayCnt) * 100.0 / max(newUsers) `3日留存率`,
         |    max(fourDayStayCnt) * 100.0 / max(newUsers) `4日留存率`,
         |    max(fiveDayStayCnt) * 100.0 / max(newUsers) `5日留存率`,
         |    max(sixDayStayCnt) * 100.0 / max(newUsers) `6日留存率`,
         |    max(sevenDayStayCnt) * 100.0 / max(newUsers) `7日留存率`,
         |  from
         |  (
         |    (select
         |      count(distinct userName) `newUser`,
         |      0 `activeUser`,
         |      0 `nextDayStayCnt`,
         |      0 `twoDayStayCnt`,
         |      0 `threeDayStayCnt`,
         |      0 `fourDayStayCnt`,
         |      0 `fiveDayStayCnt`,
         |      0 `sixDayStayCnt`,
         |      0 `sevenDayStayCnt`
         |    from tb_name
         |    where eventType='1' and(
         |                              getTimeMills(time) >= UNIX_TIMESTAMP('$baseDate',yyyy-MM-dd) * 1000 and
         |                              getTimeMills(time) < UNIX_TIMESTAMP(DATE_ADD('$baseDate',1),'yyyy-MM-dd') * 1000
         |                            )
         |    )
         |    union all
         |    (select
         |    0 `newUser`,
         |    count(distinct userName) `activeUser`,
         |    count(distinct nextDayUser) `nextDayStayCnt`,
         |    count(distinct twoDayUser) `twoDayStayCnt`,
         |    count(distinct threeDayUser) `threeDayStayCnt`,
         |    count(distinct fourDayUser) `fourDayStayCnt`,
         |    count(distinct fiveDayUser) `fiveDayStayCnt`,
         |    count(distinct sixDayUser) `sixDayStayCnt`,
         |    count(distinct sevenDayUser) `sevenDayStayCnt`
         |  from
         |      (select
         |        case when
         |             (getTimeMills(t2.time) >= UNIX_TIMESTAMP(DATE_ADD('$baseDate',1),'yyyy-MM-dd') * 1000 and
         |             getTimeMills(t2.time) < UNIX_TIMESTAMP(DATE_ADD('$baseDate',2),'yyyy-MM-dd') * 1000)
         |             then t1.userName end `nextDayUser`,
         |        case when
         |             (getTimeMills(t2.time) >= UNIX_TIMESTAMP(DATE_ADD('$baseDate',2),'yyyy-MM-dd') * 1000 and
         |             getTimeMills(t2.time) < UNIX_TIMESTAMP(DATE_ADD('$baseDate',3),'yyyy-MM-dd') * 1000)
         |             then t1.userName end `twoDayUser`,
         |        case when
         |             (getTimeMills(t2.time) >= UNIX_TIMESTAMP(DATE_ADD('$baseDate',3),'yyyy-MM-dd') * 1000 and
         |             getTimeMills(t2.time) < UNIX_TIMESTAMP(DATE_ADD('$baseDate',4),'yyyy-MM-dd') * 1000)
         |             then t1.userName end `threeDayUser`,
         |        case when
         |             (getTimeMills(t2.time) >= UNIX_TIMESTAMP(DATE_ADD('$baseDate',4),'yyyy-MM-dd') * 1000 and
         |             getTimeMills(t2.time) < UNIX_TIMESTAMP(DATE_ADD('$baseDate',5),'yyyy-MM-dd') * 1000)
         |             then t1.userName end `fourDayUser`,
         |        case when
         |             (getTimeMills(t2.time) >= UNIX_TIMESTAMP(DATE_ADD('$baseDate',5),'yyyy-MM-dd') * 1000 and
         |             getTimeMills(t2.time) < UNIX_TIMESTAMP(DATE_ADD('$baseDate',6),'yyyy-MM-dd') * 1000)
         |             then t1.userName end `fiveDayUser`,
         |        case when
         |             (getTimeMills(t2.time) >= UNIX_TIMESTAMP(DATE_ADD('$baseDate',6),'yyyy-MM-dd') * 1000 and
         |             getTimeMills(t2.time) < UNIX_TIMESTAMP(DATE_ADD('$baseDate',7),'yyyy-MM-dd') * 1000)
         |             then t1.userName end `sixDayUser`,
         |        case when
         |             (getTimeMills(t2.time) >= UNIX_TIMESTAMP(DATE_ADD('$baseDate',7),'yyyy-MM-dd') * 1000 and
         |             getTimeMills(t2.time) < UNIX_TIMESTAMP(DATE_ADD('$baseDate',8),'yyyy-MM-dd') * 1000)
         |             then t1.userName end `sevenDayUser`
         |      from
         |        tb_game t1, -- t1:存储的是基准日的用户日志信息
         |        tb_game t2  -- t2:存储的是次日~七日的用户的日志信息
         |      where t1.userName=t2.userName -- 去除笛卡尔积
         |          and t2.eventType='2' -- 次日的事件类型只是登录
         |          and t1.eventType='1' -- 基准日的事件类型只是注册
         |          and getTimeMills(t1.time) >= UNIX_TIMESTAMP('$baseDate', 'yyyy-MM-dd') * 1000
         |          and getTimeMills(t1.time) < UNIX_TIMESTAMP(DATE_ADD('$baseDate',1), yyyy-MM-dd) * 1000 --基准日的时间
         |    )
         |  )
         |)
         |""".stripMargin)
      .rdd.first

    val date = baseDate
    val newAddUserCnt: Long = row.getAs[Long]("新增用户数")
    val activeUserCnt: Long = row.getAs[Long]("活跃用户数")
    val nextDayRate: String = s"${row.getAs[java.math.BigDecimal]("次日留存率")}%.2f".concat("%")
    val twoDayRate: String = s"${row.getAs[java.math.BigDecimal]("2日留存率")}%.2f".concat("%")
    val threeDayRate: String = s"${row.getAs[java.math.BigDecimal]("3日留存率")}%.2f".concat("%")
    val fourDayRate: String = s"${row.getAs[java.math.BigDecimal]("4日留存率")}%.2f".concat("%")
    val fiveDayRate: String = s"${row.getAs[java.math.BigDecimal]("5日留存率")}%.2f".concat("%")
    val sixDayRate: String = s"${row.getAs[java.math.BigDecimal]("6日留存率")}%.2f".concat("%")
    val sevenDayRate: String = s"${row.getAs[java.math.BigDecimal]("7日留存率")}%.2f".concat("%")

    //6.将计算后的结果落地到RDBMS中固化起来
    //6.1Dao实例准备
    val dao: GameAnaylysisResultImpl = new GameAnaylysisResultImpl

    //6.2构建实例
    val bean: GameAnaylysisResultBean = new GameAnaylysisResultBean(date,
      newAddUserCnt,
      activeUserCnt,
      nextDayRate,
      twoDayRate,
      threeDayRate,
      fourDayRate,
      fiveDayRate,
      sixDayRate,
      sevenDayRate
    )

    //save
    dao.save(bean)
    //关闭
    spark.stop()
  }

  /**
   * 处理时间,形如,yyyy年MM月dd日,E,HH:mm:ss, 如:2018年2月1日,星期一,10:02:05
   * @param time
   * @return
   */
  private def processTime(time: String) = {
    val sdf = new SimpleDateFormat(CommonUtils.getPropertiesValueBykey(CommonData.TIME_PATTERN2))
    val date = sdf.parse(time)
    val baseDateMillis: Long = date.getTime
    //返回
    baseDateMillis
  }

  /**
   * 从es分布式集群中读取数据,并对数据进行标准化,清洗掉脏数据,将RDD转换为DataFrame
   * 然后将DataFrame映射为一张虚拟表,然后缓存起来
   * @param spark
   * @param sc
   */
  private def readDataFromES(spark: SparkSession, sc: SparkContext) = {
    //从ES中读取数据,并cache
    import org.elasticsearch.spark._
    //SparkSQL相关
    import spark.implicits._
    val query =
      """
        |{
        |    "query":{
        |     "match_all":  {}
        |    }
        |}
        |""".stripMargin

    //调查:是指索引库中的全部数据,还是一页的十条数据
//    val cnt = sc.esRDD("gamelog", query).count()
//    println(s"cnt=$cnt")

    val frame: DataFrame = sc.esRDD("gamelog", query)
      .map(perEle => {
        val record = perEle._2
        //用户名
        val userName = record.getOrElse("userName", "").asInstanceOf[String]
        //时间类型
        val eventType = record.getOrElse("event_type", "").asInstanceOf[String]
        //玩游戏的时间
        val time = record.getOrElse("time", "").asInstanceOf[String]
        val ip = record.getOrElse("ip", "").asInstanceOf[String]
        (userName, eventType, time, ip)
      }) //从es中筛选出需要的field
      .filter(perEle => { //将没有ip记录的信息过滤
        val ip = perEle._4
        val regex = """(\d{1,3}\.){3}\d{1,3}""" //若是正则表达式包含的特殊符号,使用三双引号,不用转义
        ip.matches(regex)
      }).toDF("userName", "eventType", "time", "ip")

    //映射为虚拟表
    frame.createOrReplaceTempView("tb_game")
    //缓存表
    spark.sqlContext.cacheTable("tb_game")
  }

  /**
   * 获得SparkSession和SparkContext
   * @return
   */
  private def getSparkContext = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local[*]")
      .appName(this.getClass.getSimpleName)
      .config("es.nodes", "node01,node02,node03")
      .config("port", "9200")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext
    (spark,sc)
  }
}

















